package com.atguigu.sql;

import com.atguigu.beans.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author Felix
 * @date 2024/3/4
 *  该案例演示了
 *      流->表->持续查询->表->流
 */
public class Flink02_Stream2Table2Stream {
    public static void main(String[] args) throws Exception {
        //准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //准备一条流
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 2L, 2),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s3", 4L, 4)
        );

        //将流转换为动态表
        tableEnv.createTemporaryView("t_sensor",sensorDS);

        //持续查询---结果还是动态表
        //Table resTable = tableEnv.sqlQuery("select * from t_sensor");
        Table resTable = tableEnv.sqlQuery("select id,sum(vc) as sumVc from t_sensor group by id");
        //将动态表转换为流
        //如果是追加查询，可以直接调用toDataStream方法
        //DataStream<Row> ds = tableEnv.toDataStream(resTable);
        //DataStream<WaterSensor> ds = tableEnv.toDataStream(resTable, WaterSensor.class);
        //如果是更新查询，可以直接调用toChangelogStream方法
        DataStream<Row> ds = tableEnv.toChangelogStream(resTable);

        ds.print();

        env.execute();

    }
}
